/*
 * @Author: Wangjun
 * @Date: 2023-08-09 09:41:36
 * @LastEditTime: 2023-10-26 17:04:28
 * @LastEditors: Wangjun
 * @Description:kafka 消费者
 * @FilePath: \xr_node_calc\kafka\reader.go
 * hnxr
 */
package kafka

import (
	"strings"

	"github.com/twmb/franz-go/pkg/kgo"

	"gitee.com/haodreams/golib/logs"
	"gitee.com/haodreams/libs/config"
)

func NewReader(topic string) (kafkaC *kgo.Client) {
	host := config.String("kafka.hosts.consumer")
	brokers := strings.Split(host, ",")
	groupID := "xr_node_task_" + config.String("kafka.group.consumer")

	opts := []kgo.Opt{
		kgo.SeedBrokers(brokers...),
		kgo.ConsumerGroup(groupID),
		kgo.ConsumeTopics(topic),
		//kgo.ConsumeResetOffset(kgo.NewOffset().AtEnd()),
	}
	//关闭自动提交
	opts = append(opts, kgo.DisableAutoCommit())

	cl, err := kgo.NewClient(opts...)
	if err != nil {
		logs.Warn(err)
		return nil
	}
	return cl
}

func NewReaderNode(topic, host string) (kafkaC *kgo.Client) {
	brokers := strings.Split(host, ",")
	groupID := config.String("kafka.group.consumer") + "_" + "op_ticker"

	opts := []kgo.Opt{
		kgo.SeedBrokers(brokers...),
		kgo.ConsumerGroup(groupID),
		kgo.ConsumeTopics(topic),
		//kgo.ConsumeResetOffset(kgo.NewOffset().AtEnd()),
	}
	//关闭自动提交
	opts = append(opts, kgo.DisableAutoCommit())

	cl, err := kgo.NewClient(opts...)
	if err != nil {
		logs.Warn(err)
		return nil
	}
	return cl
}
